package org.myorg.kafka2mysql.demo.kafka;


import com.alibaba.fastjson.JSON;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.shaded.curator.org.apache.curator.shaded.com.google.common.collect.Lists;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.PrintSinkFunction;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer011;
import org.apache.flink.util.Collector;
import org.myorg.kafka2mysql.demo.entitys.Student;
import org.myorg.kafka2mysql.demo.mysql.SinkListToMysql;

import java.util.ArrayList;
import java.util.List;
import java.util.Properties;

public class KafkaConsumer2Mysql {

	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		Properties props = new Properties();
		props.put("bootstrap.servers", "172.18.1.87:9092");
		//		props.put("zookeeper.connect", "172.18.1.87:2181");
		props.put("group.id", "metric-group");
		props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
		props.put("auto.offset.reset", "latest");

		SingleOutputStreamOperator<Student> student = env.addSource(
				new FlinkKafkaConsumer011<>(
						KafkaDsProducer.topic,
						new SimpleStringSchema(), // String 序列化
						props
				)
		).setParallelism(1).map(string -> JSON.parseObject(string, Student.class));

		SingleOutputStreamOperator<List<Student>> apply = student.timeWindowAll(Time.seconds(10)).apply(new AllWindowFunction<Student, List<Student>, TimeWindow>() {
			@Override
			public void apply(TimeWindow timeWindow, Iterable<Student> iterable, Collector<List<Student>> collector) throws Exception {
				ArrayList<Student> list = Lists.newArrayList(iterable);
				if (list.size() > 0) {
					System.out.println("10s收集到的 student 数据条数是：" + list.size());
					collector.collect(list);
				}
			}
		});

		apply.addSink(new SinkListToMysql());
		apply.addSink(new PrintSinkFunction<>());

		env.execute("Flink add mysql sink");
	}
}
